package com.reactim.message.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.reactim.message.controller.MessageRSocketController;
import com.reactim.message.dto.MessageResponse;
import com.reactim.message.service.impl.MessagePushServiceImpl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 消息推送监听器
 * 监听Kafka消息队列，处理消息推送
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MessagePushListener {
    
    private final MessageRSocketController rSocketController;
    private final ObjectMapper objectMapper;
    
    /**
     * 监听消息推送队列
     */
    @KafkaListener(topics = "message-push", groupId = "message-service")
    public void handleMessagePush(String message) {
        try {
            MessagePushServiceImpl.PushMessage pushMessage = 
                objectMapper.readValue(message, MessagePushServiceImpl.PushMessage.class);
            
            switch (pushMessage.getType()) {
                case NEW_MESSAGE:
                    handleNewMessage(pushMessage);
                    break;
                case MESSAGE_RECALL:
                    handleMessageRecall(pushMessage);
                    break;
                case MESSAGE_DELETE:
                    handleMessageDelete(pushMessage);
                    break;
                case MESSAGE_EDIT:
                    handleMessageEdit(pushMessage);
                    break;
                case READ_RECEIPT:
                    handleReadReceipt(pushMessage);
                    break;
                case TYPING_INDICATOR:
                    handleTypingIndicator(pushMessage);
                    break;
                case USER_STATUS_CHANGE:
                    handleUserStatusChange(pushMessage);
                    break;
                default:
                    log.warn("未知的推送消息类型: {}", pushMessage.getType());
            }
            
        } catch (Exception e) {
            log.error("处理消息推送失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 监听离线通知队列
     */
    @KafkaListener(topics = "offline-notification", groupId = "message-service")
    public void handleOfflineNotification(String message) {
        try {
            MessagePushServiceImpl.OfflineNotification notification = 
                objectMapper.readValue(message, MessagePushServiceImpl.OfflineNotification.class);
            
            // 处理离线通知（发送推送通知、邮件、短信等）
            processOfflineNotification(notification);
            
        } catch (Exception e) {
            log.error("处理离线通知失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 处理新消息推送
     */
    private void handleNewMessage(MessagePushServiceImpl.PushMessage pushMessage) {
        Long userId = pushMessage.getUserId();
        MessageResponse messageResponse = (MessageResponse) pushMessage.getData();
        
        // 通过RSocket推送给在线用户
        rSocketController.pushMessageToUser(userId, messageResponse)
                .subscribe(
                    null,
                    error -> log.error("RSocket推送失败: userId={}, error={}", userId, error.getMessage()),
                    () -> log.debug("RSocket推送成功: userId={}, messageId={}", userId, messageResponse.getMessageId())
                );
    }
    
    /**
     * 处理消息撤回推送
     */
    private void handleMessageRecall(MessagePushServiceImpl.PushMessage pushMessage) {
        Long userId = pushMessage.getUserId();
        
        // 构建撤回通知
        MessageRecallNotification notification = MessageRecallNotification.builder()
                .type("MESSAGE_RECALL")
                .data(pushMessage.getData())
                .timestamp(pushMessage.getTimestamp())
                .build();
        
        // 推送撤回通知 - 这里需要创建一个通用的推送方法
        // rSocketController.pushNotificationToUser(userId, notification)
        //         .subscribe();
        log.debug("消息撤回通知: userId={}", userId);
    }
    
    /**
     * 处理消息删除推送
     */
    private void handleMessageDelete(MessagePushServiceImpl.PushMessage pushMessage) {
        // 类似消息撤回的处理逻辑
        log.debug("处理消息删除推送: userId={}", pushMessage.getUserId());
    }
    
    /**
     * 处理消息编辑推送
     */
    private void handleMessageEdit(MessagePushServiceImpl.PushMessage pushMessage) {
        // 推送编辑后的消息
        handleNewMessage(pushMessage);
    }
    
    /**
     * 处理已读回执推送
     */
    private void handleReadReceipt(MessagePushServiceImpl.PushMessage pushMessage) {
        Long userId = pushMessage.getUserId();
        
        // 构建已读回执通知
        ReadReceiptNotification notification = ReadReceiptNotification.builder()
                .type("READ_RECEIPT")
                .data(pushMessage.getData())
                .timestamp(pushMessage.getTimestamp())
                .build();
        
        // TODO: 实现已读回执通知推送
        log.debug("已读回执通知: userId={}", userId);
    }
    
    /**
     * 处理正在输入指示器
     */
    private void handleTypingIndicator(MessagePushServiceImpl.PushMessage pushMessage) {
        Long userId = pushMessage.getUserId();
        
        // 构建输入指示器通知
        TypingIndicatorNotification notification = TypingIndicatorNotification.builder()
                .type("TYPING_INDICATOR")
                .data(pushMessage.getData())
                .timestamp(pushMessage.getTimestamp())
                .build();
        
        // TODO: 实现输入指示器通知推送
        log.debug("输入指示器通知: userId={}", userId);
    }
    
    /**
     * 处理用户状态变更推送
     */
    private void handleUserStatusChange(MessagePushServiceImpl.PushMessage pushMessage) {
        Long userId = pushMessage.getUserId();
        
        // 构建状态变更通知
        UserStatusNotification notification = UserStatusNotification.builder()
                .type("USER_STATUS_CHANGE")
                .data(pushMessage.getData())
                .timestamp(pushMessage.getTimestamp())
                .build();
        
        // TODO: 实现用户状态变更通知推送
        log.debug("用户状态变更通知: userId={}", userId);
    }
    
    /**
     * 处理离线通知
     */
    private void processOfflineNotification(MessagePushServiceImpl.OfflineNotification notification) {
        log.info("处理离线通知: userId={}, messageId={}, content={}", 
            notification.getUserId(), notification.getMessageId(), notification.getContent());
        
        // TODO: 实现具体的离线通知逻辑
        // 1. 发送推送通知（APNs、FCM等）
        // 2. 发送邮件通知
        // 3. 发送短信通知
        // 4. 更新未读消息计数
        
        // 示例：发送推送通知
        sendPushNotification(notification);
    }
    
    /**
     * 发送推送通知
     */
    private void sendPushNotification(MessagePushServiceImpl.OfflineNotification notification) {
        // TODO: 集成推送服务（如极光推送、个推等）
        log.debug("发送推送通知: userId={}, content={}", 
            notification.getUserId(), notification.getContent());
    }
    
    // 通知数据结构
    @lombok.Data
    @lombok.Builder
    @lombok.NoArgsConstructor
    @lombok.AllArgsConstructor
    private static class MessageRecallNotification {
        private String type;
        private Object data;
        private Long timestamp;
    }
    
    @lombok.Data
    @lombok.Builder
    @lombok.NoArgsConstructor
    @lombok.AllArgsConstructor
    private static class ReadReceiptNotification {
        private String type;
        private Object data;
        private Long timestamp;
    }
    
    @lombok.Data
    @lombok.Builder
    @lombok.NoArgsConstructor
    @lombok.AllArgsConstructor
    private static class TypingIndicatorNotification {
        private String type;
        private Object data;
        private Long timestamp;
    }
    
    @lombok.Data
    @lombok.Builder
    @lombok.NoArgsConstructor
    @lombok.AllArgsConstructor
    private static class UserStatusNotification {
        private String type;
        private Object data;
        private Long timestamp;
    }
}